package fnic.prehand.esagent.passitive_measure;
 
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.Iterator;
import java.util.List;

import org.apache.log4j.Logger;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.common.xcontent.json.JsonXContentGenerator;

import fnic.prehand.augment.Flow;
import fnic.prehand.esagent.IndexAgent;

public class PassitiveAgent extends IndexAgent implements Runnable{
	private static final Logger log = Logger.getLogger(PassitiveAgent.class);
	//被动
	private static final List<String> flowFieldNames;
	private static final List<String> augmentFieldNames;

//	private static final int fieldsize;
	private JsonXContentGenerator jsongenerator;
	private ByteArrayOutputStream outstream;
//	private MakeAugment augmentmaker; //扩维
//	private Aggregate aggrResults;
	
//	private List<String> records;
	Iterator<Flow> flowIterator;
	private ObjectPool agentPool;

	static{
        flowFieldNames = config.getPflowfieldname();
        augmentFieldNames = config.getPaugfieldname();
		if(flowFieldNames.size() != MakeAugment.getFlowFieldNum() || augmentFieldNames.size() != MakeAugment.getAugmentFieldNum()){
			throw new IllegalArgumentException("[p.redis.flow-augment.field] illegal!");
		}
//		fieldsize = flowFieldNames.size();
	}
	
//	public PassitiveAgent(String idx, String type, Aggregate aggr) {
	public PassitiveAgent(String idx, String type, ObjectPool agentPool) {
		super(idx, type);
		this.agentPool = agentPool;
		// TODO Auto-generated constructor stub
		outstream = new ByteArrayOutputStream();
		try {
			jsongenerator = (JsonXContentGenerator) jsoncontent.createGenerator(outstream);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			log.error(e.getMessage());
		}
//		augmentmaker = new MakeAugment();
//		aggrResults = aggr;
	}

	@Override
	protected void dealBulkIndexRsp(BulkResponse response) {
		//打印失败, 暂时不需要处理
		if(response.hasFailures()){
			log.error("failed to execute" + response.buildFailureMessage());
		}
	}

	@Override
	protected String jsonAugment(Object flowcsv) {
//		if(flowdata == null){
//			logger.error("make augment fail, flow record's field_num != 19");
//			return null;
//		}
		Flow flowdata = (Flow) flowcsv;
//		augmentmaker.augment(flowdata);
		
		String json;
		try {
			jsongenerator.writeStartObject();

			for(int i=0; i<flowdata.num_oldFlow.length; i++){
				jsongenerator.writeNumberField(flowFieldNames.get(i), flowdata.num_oldFlow[i]);
			}
			for(int i=0; i<flowdata.augFlow.length; i++){
				jsongenerator.writeStringField(augmentFieldNames.get(i), flowdata.augFlow[i]);
			}
			
			jsongenerator.writeEndObject();
			jsongenerator.flush();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			log.error(e.getMessage());
		}
		

		json = outstream.toString();
		outstream.reset();
		return json;
	}

	/*
	 * 将当前对象currentRequest队列中的数据建立索引
	 * 调用时机由业务进行控制
	 */
//	public void doBulkIndex() {
//		int actionCount = currentRequest.numberOfActions();
//		if (!quickNeedIndex || actionCount == 0) {
//			return;
//		}
//		sessionSize += actionCount;
//		try {
//			BulkResponse response = currentRequest.execute().actionGet();
//			dealBulkIndexRsp(response);
//		} catch (Exception e) {
//			log.error("Failed to process bulk", e);
//		}
//		log.info("one bulk , current size=" + sessionSize);
//	}
	
	/*
	 * 重新设置currentRequest队列
	 * 调用时机由业务进行控制
	 */
//	public void resetCurRequest(){
//		quickNeedIndex = false;
//		currentRequest = client.prepareBulk();
//	}
	public void setFlowIterator(Iterator<Flow> flowIterator) {
		this.flowIterator = flowIterator;
	}

	public void run() {
		while(flowIterator.hasNext()){
			atomProcess(flowIterator.next());
		}
		processBulkIfNeeded(true);
		agentPool.returnObject(this);
	}

}
